package com.liang.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.ConfirmListener;
import sun.applet.Main;
import utils.RabbitmqUtil;

import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * User: XIAOLIANG
 * Date: 2021/6/17
 * Time: 23:56
 * Description:
 *
 * 发布确认模式
 *
 * 1 单个确认
 * 2 批量确认
 * 3 异步批量确认
 *
 *
 */
public class ConfirmMessage {


    public static final String QUEUE_NAME = "confirm_queue";

    public static final int MESSAGES_COUNT = 1000;

//
//     * 1 单个确认
// * 2 批量确认
// * 3 异步批量确认


//    单个确认
    public static void publishMessageSingle() throws IOException, TimeoutException, InterruptedException {

        Channel channel = RabbitmqUtil.getChannel();

        //队列的声明
        channel.queueDeclare(QUEUE_NAME+"1", true,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        //大量的发消息

        for (int i = 0; i < MESSAGES_COUNT; i++) {

            String message = i+"";

            channel.basicPublish("",QUEUE_NAME+"1",null,message.getBytes());

            boolean flag = channel.waitForConfirms();

            if (!flag){

//                System.out.println("消息发送成功:"+i);
                throw new RuntimeException("未收到确认的消息");
            }

        }

        System.out.println("单独确认 发布1000条消息 消耗的时间:毫秒");
        System.out.println(System.currentTimeMillis()-begin);


    }


    //批量确认
    public static void publishMessageBatch() throws IOException, InterruptedException, TimeoutException {
        Channel channel = RabbitmqUtil.getChannel();

        //队列的声明
        channel.queueDeclare(QUEUE_NAME+"2", true,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        //大量的发消息

        //批量确认消息的大小

        for (int i = 1; i <=MESSAGES_COUNT; i++) {

            String message = i+"";

            channel.basicPublish("",QUEUE_NAME+"2",null,message.getBytes());

            boolean flag = channel.waitForConfirms();

            //判断 达到100 条确认一次
            if (i==1000){

                System.out.println(channel.waitForConfirms());


            }

        }

        System.out.println("批量确认 发布1000条消息 消耗的时间:毫秒");
        System.out.println(System.currentTimeMillis()-begin);


    }


    //异步发布确认

    public static void publishMessageAsync() throws IOException, TimeoutException {

        Channel channel = RabbitmqUtil.getChannel();

        channel.queueDeclare("messageAsync",true,false,false,null);


        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         *
         * 1 轻松的将序号 与消息进行关联
         *
         * 2 轻松的进行批量的删除 条目 ,通过序号删除消息
         *
         * 3 支持高并发
         *
         */

        ConcurrentSkipListMap<Long,String> concurrentSkipListMap = new ConcurrentSkipListMap<Long,String>();


//        final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());





        //开启发布确认
        channel.confirmSelect();

        //记录时间
        long beginTime = System.currentTimeMillis();

        //准备一个消息的监听器  监听那些成功 那些失败


        //监听成功的消息 的回调函数
//        ConfirmCallback ackCallback = new ConfirmListener() {
//
//
//            @Override
//            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
//
//            }
//
//            @Override
//            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
//
//            }
//        };
//
//
//        //监听 投递失败的消息 的回调函数
//        ConfirmCallback nackCallback = new ConfirmCallback() {
//            @Override
//            public void handle(long deliveryTag, boolean multiple) throws IOException {
//                String message =  concurrentSkipListMap.get(deliveryTag);
//                System.out.println("未确认的消息:"+message);
//
//            }
//        };

        //监听器 准备完成
        channel.addConfirmListener(new ConfirmListener(){

            //
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                if (multiple) {
//                    System.out.println("handleNack : " + arg1);
                    concurrentSkipListMap.headMap(deliveryTag + 1).clear();
                } else {
//                    System.out.println("handleNack : " + arg1);
                    concurrentSkipListMap.remove(deliveryTag);
                }


            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {

                System.out.println(
                        "消息投递失败:"+concurrentSkipListMap.get(deliveryTag)
                );

            }
        });


        for (int i = 1; i <= 1000; i++) {

            channel.basicPublish("","messageAsync",null,("消息:"+i).getBytes());

            //记录下 所有要发送的消息
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),"消息"+i);
        }

        System.out.println("1000条异步确认消息 耗费的时间:"+(System.currentTimeMillis()-beginTime));

        try {
            Thread.sleep(8*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("剩余的消息:"+concurrentSkipListMap.size());

        for (Map.Entry<Long, String> longStringEntry : concurrentSkipListMap.entrySet()) {

            System.out.println(longStringEntry.getKey()+":"+longStringEntry.getValue());

        }


    }



    public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {


//        publishMessageSingle();
//        publishMessageBatch();
        publishMessageAsync();

    }


}